package march.finalmarch21.march03;

import march.finalmarch21.FlowBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;

public class March03 extends Configured implements Tool {
    public static Map<String, BigDecimal> huiLvMap = new HashMap<>();

    static class March03Mapper extends Mapper<LongWritable, Text,Text,Text>{
        String fileName;


        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            fileName = inputSplit.getPath().getName();

        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] split = value.toString().split(",");

            if(fileName.contains("huilv")){
                //汇率表，使用map进行存储，方便Reduce阶段取
                huiLvMap.put(split[0],new BigDecimal(split[1]));
            }else{
                //获取项目类别，并以项目类别作为key传入Reduce
                //利用了Reduce去重的特性
                String type = split[2];

                context.write(new Text(type),value);
            }



        }
    }

    static class March03Reducer extends Reducer<Text,Text, NullWritable,Text>{

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            //建立两个Map，mapRMB用于后面对贷款总金额进行排序，flowBean用于求出前三贷款金额的id后，进行对象的查找
            HashMap<String, BigDecimal> mapRMB = new HashMap<>();
            HashMap<String, FlowBean> flowBeanMap = new HashMap<>();

            //对相同行业类别的value进行遍历
            //此时key为行业类别，value为每一行数据
            for (Text value : values) {
                FlowBean flowBean = new FlowBean();

                String[] split = value.toString().split(",");
                //获取项目序号、类别、币种ID、贷款金额
                String id = split[0];
                String type = split[2];
                String coinId = split[3];
                String loanAmount = split[4];

                //计算贷款金额
                //贷款金额_人民币=贷款金额*汇率表中的兑换价格

                //获取汇率
                BigDecimal huilv = huiLvMap.get(coinId);
                BigDecimal loanRMB = new BigDecimal(loanAmount).multiply(huilv);

                //填充数据
                flowBean.setId(id);
                flowBean.setType(type);
                flowBean.setCoinId(coinId);
                flowBean.setLoanAmount(loanAmount);
                flowBean.setLoanRMB(loanRMB.toString());

                //建立两个Map，mapRMB用于后面对贷款总金额进行排序，flowBean用于求出前三贷款金额的id后，进行对象的查找
                flowBeanMap.put(id,flowBean);
                mapRMB.put(id,loanRMB);
            }


            //循环三遍，取出每个最大的值，并将该key-value对给删除
            for (int i = 0; i < 3; i++) {
                String maxMapId= "";
                BigDecimal max = new BigDecimal("0.00");
                int size = 0;
                for (String id : mapRMB.keySet()) {
                    //判断目前的值，和map里的值谁大，获取大的值，一轮遍历下便可以获取最大的值
                    if(max.compareTo(mapRMB.get(id)) < 0){
                        max = mapRMB.get(id);
                        maxMapId = id;
                    }

                    //如果map的尺寸小于3，需要避免FlowBean的空指针异常
                    if(size == mapRMB.size()){
                        return;
                    }
                    size ++;
                }
                //每次取出最大值，把该key-value对删除
                mapRMB.remove(maxMapId);
                FlowBean flowBean = flowBeanMap.get(maxMapId);

                if(flowBean == null){
                    return;
                }
                //输出行业类别、币种Id、贷款金额（RMB）
                Text text = new Text(flowBean.getType()+","+flowBean.getCoinId()+","+flowBean.getLoanRMB());
                context.write(NullWritable.get(),text);
            }

        }
    }


    @Override
    public int run(String[] args) throws Exception {
        Configuration conf=getConf();

        Job job=Job.getInstance(conf,this.getClass().getSimpleName());
        job.setJarByClass(March03.class);

        job.setMapperClass(March03.March03Mapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(March03.March03Reducer.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        TextInputFormat.addInputPath(job,new Path("F:\\test\\march01Input"));
        TextOutputFormat.setOutputPath(job,new Path("F:\\test\\march01Output"));

        return job.waitForCompletion(true)?0:1;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new March03(),args));
    }
}
